package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.IOUtils;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Map;

/**
 * 通过 Phoenix 操作，将维度侧输出流的数据写到 HBase 中
 *
 * @author lvbingbing
 * @date 2022-03-25 10:57
 */
@Slf4j
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private transient Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /**
     * 拼接 update 语句，向 HBase 中写入维度数据
     * <p>
     * jsonObject 格式如下：
     * {"database":"gmall2021","data":{"tm_name":"zhangSan","id":13},"commit":true,"sink_table":"dim_base_trademark","type":"insert","table":"base_trademark","ts":1628134550}
     *
     * @param jsonObject 维度侧输出流中的数据
     * @param context    上下文环境
     * @throws Exception 写入失败
     */
    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        String sinkTable = jsonObject.getString("sink_table");
        JSONObject dataJsonObject = jsonObject.getJSONObject("data");
        StringBuilder columnNameSb = new StringBuilder();
        StringBuilder columnValueSb = new StringBuilder();
        for (Map.Entry<String, Object> entry : dataJsonObject.entrySet()) {
            String key = entry.getKey();
            columnNameSb.append(key).append(", ");
            String value = entry.getValue().toString();
            columnValueSb.append("'").append(value).append("'").append(", ");
        }
        String columnNames = columnNameSb.substring(0, columnNameSb.length() - 2);
        String columnValues = columnValueSb.substring(0, columnValueSb.length() - 2);
        StringBuilder stringBuilder = new StringBuilder();
        String upsertSql = stringBuilder.append("upsert into ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append(" (")
                .append(columnNames)
                .append(") values (")
                .append(columnValues)
                .append(")")
                .toString();
        // 创建数据库操作对象
        try (PreparedStatement ps = connection.prepareStatement(upsertSql)) {
            // 执行 SQL
            ps.executeUpdate();
            // 由于 phoenix 的连接实现类不是自动提交事务，需要手动提交事务
            connection.commit();
        } catch (Exception e) {
            log.error("向Phoenix维度表中插入数据失败！失败原因：{}", e);
            throw e;
        }
        // 先写入数据库，再失效缓存
        // 如果维度数据发生变化，清空当前数据在 redis 中的缓存
        String type = jsonObject.getString("type");
        if (type.equals("update") || type.equals("delete")) {
            String id = dataJsonObject.getString("id");
            log.info("维度数据发生变化，清除缓存！tableName：{}, id：{}", sinkTable, id);
            DimUtils.clearCache(sinkTable, id);
        }
    }

    @Override
    public void close() throws Exception {
        IOUtils.closeQuietly(connection);
    }
}